package skewreduce.dfof;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.logical.LFinalizeOp;
import skewreduce.framework.logical.LMergeOp;
import skewreduce.framework.logical.LPartitionOp;
import skewreduce.framework.logical.LSampleOp;
import skewreduce.framework.logical.LProcessOp;
import skewreduce.framework.physical.PPigOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.framework.physical.UniformSample;
import skewreduce.lib.Cube2;
import skewreduce.lib.Partition;

public class dFoF extends EaggDriver {
    private boolean newMerge;

    public String getAppName() { return "dFoF"; }

    public LSampleOp createLSampleOp() {
    	if ( getConf().getBoolean("dfof.sample.uniform",true) )
    		return UniformSample.createLogicalOp(this);
    	
    	return new LSampleOp(this) {
    		@Override
    		protected PhysicalOp createInstance() {
    			return new dFoFSample();
    		}
		};
    }
    
    public LPartitionOp createLPartitionOp() {
        return new LPartitionOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new dFoFPartition();
            }
        };
    }

    public LProcessOp createLWorkOp() {
        return new LProcessOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new LocalFoF();
            }
        };
    }

    public LMergeOp createLMergeOp() {
        return new LMergeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                boolean mapOnly = getConfiguration().getBoolean("skewreduce.merge.mapOnly",false);
                if ( mapOnly ) {
                    return new dFoFMerge3();
                } else {
                    return ( newMerge ) ? new dFoFMerge2() : new dFoFMerge();
                }
            }
        };
    }

    public LFinalizeOp createLFinalizeOp() {
        return new LFinalizeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new PPigOp(); // FIXME
            }
        };
    }

    @Override
    protected int handleOption(String[] args,int i) {
        if ( "-numPart".equals(args[i]) ) {
            if ( i+1 < args.length ) {
                int n = Integer.parseInt(args[++i]);
                System.err.println("Overriding initial number of partitions to "+n);
                getConf().setInt("skewreduce.partition.initialpartitions",n);
                return 1;
            } else {
                System.err.println("missing -numPart parameter");
            }
        } else if ( "-newMerge".equals(args[i]) ) {
            newMerge = true;
            return 0;
        }
        return -1;
    }

	@Override
	public Partition getRootPartition() {
		return new Cube2(0,1.0f);
	}
}
